package com.camemax.controller;

import com.camemax.operators.AdAggregateFunction;
import com.camemax.operators.AdWindowFunction;
import com.camemax.operators.BlackListKeyedProcessFunction;
import com.camemax.pojo.AdvertisementView;
import com.camemax.pojo.AdvertisementViewCount;
import com.camemax.pojo.BlackList;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import scala.Tuple2;

import java.net.URL;
/*
* 统计各省份广告点击量
* */
public class AdvertisementCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        URL resource = AdvertisementCount.class.getResource("/AdClickLog.csv");
        DataStream<String> fileDataStream = env.readTextFile(resource.getPath());

        DataStream<AdvertisementView> windowStream = fileDataStream.map((MapFunction<String, AdvertisementView>) line -> {
            String[] fields = line.split(",");
            return new AdvertisementView(
                    Long.valueOf(fields[0]),
                    Long.valueOf(fields[1]),
                    fields[2],
                    fields[3],
                    fields[4]
            );
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<AdvertisementView>noWatermarks()
                .withTimestampAssigner(
                        (SerializableTimestampAssigner<AdvertisementView>) (element, recordTimestamp) ->
                                Long.parseLong(element.getTimestamp()) * 1000L));

        SingleOutputStreamOperator<AdvertisementView> blackListDataStream =
                windowStream.keyBy( new KeySelector<AdvertisementView, Tuple2<Long, Long>>(){

                    @Override
                    public Tuple2<Long, Long> getKey(AdvertisementView value) throws Exception {
                        return new Tuple2<>(value.getUserId(),value.getAdvertisementId());
                    }
                })
                // 实现黑名单功能
                .process(new BlackListKeyedProcessFunction(100));

        SingleOutputStreamOperator<AdvertisementViewCount> resultDataStream = blackListDataStream
                .keyBy(AdvertisementView::getProvinceName)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .aggregate(new AdAggregateFunction(), new AdWindowFunction());

        resultDataStream.print();
        // 侧输出流输出 "黑名单" 数据
        blackListDataStream.getSideOutput(new OutputTag<BlackList>("black-list"){}).print("black-list");

        env.execute();
    }
}
